package cn.wanda.utils.db

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types._

import scala.collection.mutable.ArrayBuffer

class OracleDBManager {

    def registTmpFromOracle(sparkSessionContext: SparkSession, sql: String, url: String, username: String, password: String, tempTableName: String) {
        //从oracle 数据库表中读取数据到spark
        var colName = ArrayBuffer[String]()

        val jdbcdf = sparkSessionContext
          .read
          .format("jdbc")
          .option("url", url)
          .option("dbtable", sql)
          .option("user", username)
          .option("password", password)
          .option("driver", "oracle.jdbc.driver.OracleDriver")
          .load()

        jdbcdf.schema.foreach(s => {
            if (s.dataType.equals(DecimalType(38, 10)) || s.dataType.equals(DecimalType(4, 0))) {
                colName += s.name
            }
        })

        var df_int = jdbcdf
        colName.foreach(name => {
            df_int = df_int.withColumn(name, col(name).cast(FloatType)
            )
        })

        df_int.createOrReplaceTempView(tempTableName)
    }
}
